统称|程度_大数据综合项目网站流量日志数据分析系统(详细步骤和代码)

作者：苗淑香哈哈_405_408 | 来源：互联网 | 2023-08-16 16:28

篇首语：本文由编程笔记#小编为大家整理，主要介绍了大数据综合项目--网站流量日志数据分析系统(详细步骤和代码)相关的知识，希望对你有一定的参考价值。文章目录

篇首语：本文由编程笔记#小编为大家整理，主要介绍了大数据综合项目--网站流量日志数据分析系统(详细步骤和代码)相关的知识，希望对你有一定的参考价值。

文章目录

前言&＃xff1a;
基本概述
- Sqoop概述
- - 什么是Sqoop
- Flume概述
- - 什么是Flume
  - 为什么需要flume
- HIve概述
- - 什么是Hive
系统背景&＃xff1a;
模块开发
- 数据采集
- - 使用Flume搭建日志采集系统
- 数据预处理
- - 实现数据预处理
- 数据仓库开发
- - 数据导出
  - 日志分析系统报表展示

前言&＃xff1a;

提示&＃xff1a;这里简述我使用的版本情况&＃xff1a;
ubuntu16.04
hbase1.1.5
hive1.2.1
sqoop1.4.6
flume1.7.0
项目所使用的参考文档和代码资源和部分数据
网盘链接&＃xff1a;链接&＃xff1a;https://pan.baidu.com/s/1TIKHMBmEFPiOv48pxBKn2w
提取码&＃xff1a;0830

基本概述

为更好的理解项目架构&＃xff0c;对项目使用的一些服务补充一些基本概述&＃xff1a;

Sqoop概述

什么是Sqoop

Sqoop是Apache旗下的一款开源工具&＃xff0c;2013年独立成为Apache的一个顶级开源项目
Sqoop主要用于在Hadoop和关系数据库或大型机之间传输数据&＃xff0c;可以使用Sqoop工具将数据从关系数据库管理系统导入&＃xff08;import&＃xff09;到Hadoop分布式文件系统中&＃xff0c;或者将Hadoop中的数据转换导出&＃xff08;export&＃xff09;到关系数据库管理系统&＃xff0c;功能图如下

目前sqoop主要分为sqoop1和sqoop2两个版本&＃xff0c;其中版本号为1.4.x属于sqoop1&＃xff0c;而版本号为1.99.x的属于Sqoop2.这两个版本开发时的定位方向不同&＃xff0c;体系结构具有很大的差异&＃xff0c;因此它们之间互不兼容。
Sqoop1功能结构简单&＃xff0c;部署方便。提供命令行操作方式&＃xff0c;主要适用于系统服务管理人员进行简单的数据迁移操作&＃xff0c;该项目只用到sqoop1解决数据迁移问题&＃xff0c;因此我们使用sqoop1就可以完成基本的需求

Flume概述

什么是Flume

Flume原是Cloudera公司提供的一个高可用、高可靠、分布式海量日志采集、聚合和传输系统&＃xff0c;后来纳入到Apache旗下&＃xff0c;作为一个顶级开源项目。Apache Flume 不仅只限于日志数据的采集&＃xff0c;由于Flume采集的数据源是可定制的&＃xff0c;因此Flume还课用于传输大量事件数据&＃xff0c;包括但不限于网络流量数据、社交媒体生成的数据以及几乎任何可能的数据源
当前Flume分成两个版本&＃xff1a;Flume0.9x统称Flume-og和Flume1.x版本&＃xff0c;统称Flume-ng&＃xff0c;早期Flume-og存在设计不合理&＃xff0c;纳入Apache旗下后对Flume代码进行重构&＃xff0c;进行补充和加强该项目也是使用Flume-ng版本进行Flume开发
Flume运行机制

Agent&＃xff1a;Agent是Flume中的核心组件&＃xff0c;用来收集数据。一个Agent就是一个JVM进程&＃xff0c;它是Flume中最小的独立运行的单元。

Flume的核心是把数据从数据源&＃xff08;如Web Server&＃xff09;通过数据采集器&＃xff08;Source&＃xff09;收集过来&＃xff0c;
再将收集的数据通过缓冲通道&＃xff08;Channel&＃xff09;汇集到指定的接收器&＃xff08;Sink&＃xff09;。

为什么需要flume

1、当大量的数据在同一个时间要写入HDFS时&＃xff0c;每次一个文件被创建或者分配一个新的块&＃xff0c;都会在namenode发生很复杂的操作&＃xff0c;主节点压力很大&＃xff0c;会造成很多问题&＃xff0c;比如写入时间严重延迟、写入失败等。
2、flume是一个灵活的分布式系统&＃xff0c;易扩展&＃xff0c;高度可定制化。
3、flume中的核心组件Agent。一个Agent可以连接一个或者多个Agent&＃xff0c;可以从一个或者多个Agent上收集数据。多个Agent相互连接&＃xff0c;可以建立流作业&＃xff0c;在Agent链上&＃xff0c;就能将数据从一个位置移动到另一个地方&＃xff08;HDFS、HBase等&＃xff09;。

HIve概述

Hive起源于Facebook&＃xff0c;facebook公司有着大量的日志数据&＃xff0c;而Hadoop是一个实现了Mapreduce模式开源的分布式并行计算框架。可以轻松处理大规模的数据量&＃xff0c;Mapreduce程序虽然对于熟悉Java语言的工程师来说比较容易开发&＃xff0c;但是对于其他语言使用者来说难度较大&＃xff0c;为此Facebook开发团队想到设计一种使用Sql语言就能够对日志数据查询分析的工具&＃xff0c;大大节省开发人员的学习成本&＃xff0c;HIve则诞生于此

什么是Hive

Hive是建立在Hadoop文件系统上的数据仓库&＃xff0c;它提供了一系列工具&＃xff0c;能够对存储在HDFS中的数据进行数据提取、转换和加载&＃xff08;ETL&＃xff09;&＃xff0c;这是一种可以存储、查询和分析存储在Hadoop中的大规模数据的工具。
Hive定义了简单的类SQL查询语言&＃xff0c;可以将结构化的数据文件映射为一张数据表&＃xff0c;允许熟悉SQL的用户查询数据&＃xff0c;也允许熟悉MapReduce的开发者开发定义mapper和reducer来处理复杂的分析工作&＃xff0c;这样Hive的有事更加明显

(1)用户接口:主要分为3个,分别是CLI、JDBC/ODBC 和 WebUI。其中,CLI 即 Shell终端命令行,它是最常用的方式。JDBC/ODBC是Hive的
Java实现,与使用传统数据库JDBC的方式类似,WebUI指的是通过浏览器访问Hive。
(2&＃xff09;跨语言服务(Thrift Server):Thrift是Facebook开发的一个软件框架,可以用来进行可扩展且跨语言的服务。Hive集成了该服务﹐能让不同
的编程语言调用Hive的接口。
(3&＃xff09;底层的驱动引擎:主要包含编译器(Compiler),优化器(Optimizer)和执行器(Executor ),它们用于完成HQL查询语句从词法分析、语法分析、编译、优化以及查询计划的生成,生成的查询计划存储在HDFS中,并在随后由MapReduce调用执行。
(4)元数据存储系统(Metastore):Hive中的元数据通常包含表名,列、分区及其相关属性,表数据所在目录的位置信息&＃xff0c;Metastore默认存在自带的 Derby数据库中。由于Derby数据库不适合多用户操作,并且数据存储目录不固定,不方便管理,因此,通常都将元数据存储在mysql数据库。

系统背景&＃xff1a;

近年来,随着社会的不断发展,人们对于海量数据的挖掘和运用越来越重视,互联网是面向全社会公众进行信息交流的平台,已成为了收集信息的最佳渠道并逐步进入传统的流通领域。同时,伴随着大数据技术的创新与应用,进一步为人们进行大数据统计分析提供了便利。
大数据信息的统计分析可以为企业决策者提供充实的依据。例如,通过对某网站日志数据统计分析,可以得出网站的日访问量,从而得出网站的欢迎程度;通过对移动APP的下载数据量进行统计分析,可以得出应用程序的受欢迎程度﹐甚至还可以通过不同维度(区域、时间段、下载方式等)进行进一步更深层次的数据分析,为运营分析与推广决策提供可靠的参照数据

为了更清晰地了解系统日志数据统计分析的流程及架构&＃xff0c;沟通一张架构图来了解一下

三个nginx可能产生大量的日志文件&＃xff0c;然后通过Flume分别采集这些日志文件&＃xff0c;然后将采集的日志文件放在HDFS上&＃xff0c;我们可以写Mapreduce程序来对采集后的日志进行预处理&＃xff0c;因为从服务器采集过来的数据格式不能满足我们的需求&＃xff0c;可能有一些脏数据&＃xff0c;我们要通过Mapreduce对这些文件进行清理&＃xff0c;把它们编程结构化的数据&＃xff0c;然后我们可以把清洗后的数据加载到Hive仓库里面。可以对清洗后的数据进行数据分析&＃xff0c;这一步是大数据中最重要的工作.分析之后得出我们的结果&＃xff0c;可以通过sqoop这个迁移工具将我们的结果导出到mysql&＃xff0c;导出到mysql之后我们可以通过web技术对我们的结果进行一个可视化的展示
在整个流程中&＃xff0c;系统的数据分析并不是一次性大的&＃xff0c;而是按照一定频率反复计算&＃xff0c;因而整个处理链条中的各个环节需要按照一个的先后关系紧密衔接&＃xff0c;即大量任务单元的管理调度。

模块开发

数据采集

在该项目中&＃xff0c;对数据采集模块的可靠性&＃xff0c;容错能力的要求通常不会非常严苛&＃xff0c;因此使用通用的Flume日志采集框架完全可以满足数据采集的需求

使用Flume搭建日志采集系统

a1.sources&＃61;s1 a1.channels&＃61;c1 a1.sinks&＃61;k1 a1.sources.s1.type&＃61;TAILDIR a1.sources.s1.positionFile&＃61;/root/export/data/nginx/taildir_position.json a1.sources.s1.filegroups&＃61;f1 f2 a1.sources.s1.filegroups.f1&＃61;/root/export/data/nginx/test1/access.log a1.sources.s1.filegroups.f2&＃61;/root/export/data/nginx/test2/.*log.* a1.sinks.k1.type&＃61;hdfs a1.sinks.k1.hdfs.path&＃61;hdfs://192.168.80.140:9000/weblog/%Y%m%d #上传文件的前缀 a1.sinks.k1.hdfs.filePrefix &＃61; logs- #是否按照时间滚动文件夹 #a1.sinks.k1.hdfs.round &＃61;true #多少时间单位创建一个新的文件夹 #a1.sinks.k1.hdfs.roundValue &＃61; 1 #重新定义时间单位 #a1.sinks.k1.hdfs.roundUnit &＃61; hour #是否使用本地时间戳 a1.sinks.k1.hdfs.useLocalTimeStamp &＃61; true #积攒多少个Event才flush到HDFS一次 a1.sinks.k1.hdfs.batchSize &＃61; 1000 #设置文件类型&＃xff0c;可支持压缩 a1.sinks.k1.hdfs.fileType &＃61; DataStream #多久生成一个新的文件 a1.sinks.k1.hdfs.rollInterval &＃61;0 #设置每个文件的滚动大小 a1.sinks.k1.hdfs.rollSize &＃61; 10485760 #文件的滚动与Event数量无关 a1.sinks.k1.hdfs.rollCount &＃61; 0 a1.sinks.k1.hdfs.threadsPoolSize&＃61;10 a1.sinks.k1.hdfs.callTimeout&＃61;30000 a1.channels.c1.type&＃61;memory a1.channels.c1.capacity&＃61;1000 a1.channels.c1.transactionCapacity&＃61;1000 a1.sources.s1.channels&＃61;c1 a1.sinks.k1.channel&＃61;c1

我们知道通过Flume系统采集后的网站流量日志数据会汇总到HDFS上进行保存&＃xff08;这里假设保存目录为 /root/export/data/nginx/test1/access.log&＃xff09;
采集后的数据&＃xff1a;

也可以用现成的采集后的数据&＃xff0c;在网盘资源里有
采集后的数据每个字段的含义

数据预处理

在收集的日志文件中&＃xff0c;通常不能直接将日志文件进行数据分析&＃xff0c;这是因为日志文件中有许多不合法的数据&＃xff0c;要对不合法的数据进行过滤&＃xff0c;清洗出无意义的数据信息&＃xff0c;并且将原始日志中的数据格式转换成利于后续数据分析时规范的格式&＃xff0c;根据统计需求&＃xff0c;筛选出不同主题的数据

实现数据预处理

1.创建Maven项目&＃xff0c;添加相关依赖

添加相关依赖

org.apache.hadoop hadoop-common 2.7.4 org.apache.hadoop hadoop-hdfs 2.7.4 org.apache.hadoop hadoop-client 2.7.4 org.apache.hadoop hadoop-mapreduce-client-core 2.7.4 junit junit 4.13-beta-3 compile

创建JavaBean对象&＃xff0c;封装日志记录&＃xff08;WebLogBean.java&＃xff09;

package cn.itcast.mr.weblog.bean; import org.apache.hadoop.io.Writable; import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; /** * 对接外部数据的层&＃xff0c;表结构定义最好跟外部数据源保持一致 * 术语&＃xff1a; 贴源表 * &＃64;author itcast * */ public class WebLogBean implements Writable private boolean valid &＃61; true;// 判断数据是否合法 private String remote_addr;// 记录客户端的ip地址 private String remote_user;// 记录客户端用户名称,忽略属性"-" private String time_local;// 记录访问时间与时区 private String request;// 记录请求的url与http协议 private String status;// 记录请求状态&＃xff1b;成功是200 private String body_bytes_sent;// 记录发送给客户端文件主体内容大小 private String http_referer;// 用来记录从那个页面链接访问过来的 private String http_user_agent;// 记录客户浏览器的相关信息 //设置属性值 public void set(boolean valid,String remote_addr, String remote_user, String time_local, String request, String status, String body_bytes_sent, String http_referer, String http_user_agent) this.valid &＃61; valid; this.remote_addr &＃61; remote_addr; this.remote_user &＃61; remote_user; this.time_local &＃61; time_local; this.request &＃61; request; this.status &＃61; status; this.body_bytes_sent &＃61; body_bytes_sent; this.http_referer &＃61; http_referer; this.http_user_agent &＃61; http_user_agent; public String getRemote_addr() return remote_addr; public void setRemote_addr(String remote_addr) this.remote_addr &＃61; remote_addr; public String getRemote_user() return remote_user; public void setRemote_user(String remote_user) this.remote_user &＃61; remote_user; public String getTime_local() return this.time_local; public void setTime_local(String time_local) this.time_local &＃61; time_local; public String getRequest() return request; public void setRequest(String request) this.request &＃61; request; public String getStatus() return status; public void setStatus(String status) this.status &＃61; status; public String getBody_bytes_sent() return body_bytes_sent; public void setBody_bytes_sent(String body_bytes_sent) this.body_bytes_sent &＃61; body_bytes_sent; public String getHttp_referer() return http_referer; public void setHttp_referer(String http_referer) this.http_referer &＃61; http_referer; public String getHttp_user_agent() return http_user_agent; public void setHttp_user_agent(String http_user_agent) this.http_user_agent &＃61; http_user_agent; public boolean isValid() return valid; public void setValid(boolean valid) this.valid &＃61; valid; /** * 重写toString()方法&＃xff0c;使用Hive默认分隔符进行分隔&＃xff0c;为后期导入Hive表提供便利 * &＃64;return */ &＃64;Override public String toString() StringBuilder sb &＃61; new StringBuilder(); sb.append(this.valid); sb.append("\\001").append(this.getRemote_addr()); sb.append("\\001").append(this.getRemote_user()); sb.append("\\001").append(this.getTime_local()); sb.append("\\001").append(this.getRequest()); sb.append("\\001").append(this.getStatus()); sb.append("\\001").append(this.getBody_bytes_sent()); sb.append("\\001").append(this.getHttp_referer()); sb.append("\\001").append(this.getHttp_user_agent()); return sb.toString(); /** * 序列化方法 * &＃64;param in * &＃64;throws IOException */ &＃64;Override public void readFields(DataInput in) throws IOException this.valid &＃61; in.readBoolean(); this.remote_addr &＃61; in.readUTF(); this.remote_user &＃61; in.readUTF(); this.time_local &＃61; in.readUTF(); this.request &＃61; in.readUTF(); this.status &＃61; in.readUTF(); this.body_bytes_sent &＃61; in.readUTF(); this.http_referer &＃61; in.readUTF(); this.http_user_agent &＃61; in.readUTF(); /** * 反序列化方法 * &＃64;param out * &＃64;throws IOException */ &＃64;Override public void write(DataOutput out) throws IOException out.writeBoolean(this.valid); out.writeUTF(null&＃61;&＃61;remote_addr?"":remote_addr); out.writeUTF(null&＃61;&＃61;remote_user?"":remote_user); out.writeUTF(null&＃61;&＃61;time_local?"":time_local); out.writeUTF(null&＃61;&＃61;request?"":request); out.writeUTF(null&＃61;&＃61;status?"":status); out.writeUTF(null&＃61;&＃61;body_bytes_sent?"":body_bytes_sent); out.writeUTF(null&＃61;&＃61;http_referer?"":http_referer); out.writeUTF(null&＃61;&＃61;http_user_agent?"":http_user_agent);

编写MapReduce程序&＃xff0c;执行数据预处理

package cn.itcast.mr.weblog.preprocess; import cn.itcast.mr.weblog.bean.WebLogBean; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import java.io.IOException; import java.util.HashSet; import java.util.Set; /** * 处理原始日志&＃xff0c;过滤出真实请求数据&＃xff0c;转换时间格式&＃xff0c;对缺失字段填充默认值&＃xff0c;对记录标记valid和invalid */ public class WeblogPreProcess public static void main(String[] args) throws Exception Configuration conf &＃61; new Configuration(); Job job &＃61; Job.getInstance(conf); job.setJarByClass(WeblogPreProcess.class); job.setMapperClass(WeblogPreProcessMapper.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(NullWritable.class); FileInputFormat.setInputPaths(job, new Path("d:/weblog/input")); FileOutputFormat.setOutputPath(job, new Path("d:/weblog/output")); job.setNumReduceTasks(0); boolean res &＃61; job.waitForCompletion(true); System.exit(res ? 0 : 1); public static class WeblogPreProcessMapper extends Mapper // 用来存储网站url分类数据 Set pages &＃61; new HashSet(); Text k &＃61; new Text(); NullWritable v &＃61; NullWritable.get(); /** * 设置初始化方法&＃xff0c;加载网站需要分析的url分类数据&＃xff0c;存储到MapTask的内存中&＃xff0c;用来对日志数据进行过滤 * 如果用户请求的资源是以下列形式&＃xff0c;就表示用户请求的是合法资源。 */ &＃64;Override protected void setup(Context context) throws IOException, InterruptedException pages.add("/about"); pages.add("/black-ip-list/"); pages.add("/cassandra-clustor/"); pages.add("/finance-rhive-repurchase/"); pages.add("/hadoop-family-roadmap/"); pages.add("/hadoop-hive-intro/"); pages.add("/hadoop-zookeeper-intro/"); pages.add("/hadoop-mahout-roadmap/"); &＃64;Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException //获取一行数据 String line &＃61; value.toString(); //调用解析类WebLogParser解析日志数据&＃xff0c;最后封装为WebLogBean对象 WebLogBean webLogBean &＃61; WebLogParser.parser(line); if (webLogBean !&＃61; null) // 过滤js/图片/css等静态资源 WebLogParser.filtStaticResource(webLogBean, pages); k.set(webLogBean.toString()); context.write(k, v);

定义WebLogParser类用于解析读取每行日志信息&＃xff0c;并将解析结果封装为WebLogBean对象

package cn.itcast.mr.weblog.preprocess; import cn.itcast.mr.weblog.bean.WebLogBean; import java.text.ParseException; import java.text.SimpleDateFormat; import java.util.Locale; import java.util.Set; public class WebLogParser //定义时间格式 public static SimpleDateFormat df1 &＃61; new SimpleDateFormat("dd/MMM/yyyy:HH:mm:ss", Locale.US); public static SimpleDateFormat df2 &＃61; new SimpleDateFormat("yyyy-MM-dd HH:mm:ss", Locale.US); public static WebLogBean parser(String line) WebLogBean webLogBean &＃61; new WebLogBean(); //把一行数据以空格字符切割并存入数组arr中 String[] arr &＃61; line.split(" "); //如果数组长度小于等于11&＃xff0c;说明这条数据不完整&＃xff0c;因此可以忽略这条数据 if (arr.length > 11) //满足条件的数据逐个赋值给webLogBean对象 webLogBean.setRemote_addr(arr[0]); webLogBean.setRemote_user(arr[1]); String time_local &＃61; formatDate(arr[3].substring(1)); if(null&＃61;&＃61;time_local || "".equals(time_local)) time_local&＃61;"-invalid_time-"; webLogBean.setTime_local(time_local); webLogBean.setRequest(arr[6]); webLogBean.setStatus(arr[8]); webLogBean.setBody_bytes_sent(arr[9]); webLogBean.setHttp_referer(arr[10]); //如果useragent元素较多&＃xff0c;拼接useragent if (arr.length > 12) StringBuilder sb &＃61; new StringBuilder(); for(int i&＃61;11;i sb.append(arr[i]); webLogBean.setHttp_user_agent(sb.toString()); else webLogBean.setHttp_user_agent(arr[11]); if (Integer.parseInt(webLogBean.getStatus()) >&＃61; 400) // 大于400&＃xff0c;HTTP错误 webLogBean.setValid(false); if("-invalid_time-".equals(webLogBean.getTime_local())) webLogBean.setValid(false); else webLogBean&＃61;null; return webLogBean; //添加标识 public static void filtStaticResource(WebLogBean bean, Set pages) if (!pages.contains(bean.getRequest())) bean.setValid(false); //格式化时间方法 public static String formatDate(String time_local) try return df2.format(df1.parse(time_local)); catch (ParseException e) return null;

该项目使用本地模式运行&＃xff0c;只需要在编写MapReduce程序完成后&＃xff0c;在本地D&＃xff1a;/weblog/input目录中放入将要清洗的日志文件&＃xff0c;再执行程序就即可
&＃xff08;意思就是将我们刚才flume采集的文件放到input下&＃xff09;
程序执行结果
在程序指定的输出路径可以看到这些

查看part-m-000的内容

这些黑黑的是间隔符&＃xff0c;可以看出和之前flume采集的数据的差比&＃xff0c;有很多flase的数据段

数据仓库开发

数据预处理完成后&＃xff0c;就需要将MapReduce程序的输出结果文件上传至HDFS中&＃xff0c;并使用HIVE建立相应的表结构于上床的输出结果文件产生映射关系。
我们先将程序执行结果文件

上传到hadoop集群中的主服务器&＃xff0c;通过lrzsz拖拽上传到/root/weblog

这个路径可以根据自己需求修改&＃xff0c;再上传到hdfs目录中&＃xff08;如/weblog/preprocessed&＃xff09;

hadoop fs -put part-m-0000 /weblog/preprocessed

如果出现下述情况

输入hadoop dfsadmin -safemode leave即可
上传成功

启动Hive
创建数据仓库

hive > create database weblog;

使用我们刚才创建的数据库

hive > use weblog;

创建表

hive > create table ods_weblog_origin( valid string, remote_addr string, remote_user string, time_local string, request string, status string, body_bytes_sent string, http_referer string, http_user_agent string ) partitioned by(datestr string) row format delimited fields terminated by &＃39;\\001&＃39;;

加载数据到hive表
把我们数据清洗后的数据加载到表中

hive >load data inpath &＃39;/weblog/preprocessed/&＃39; overwrite into table ods_weblog_origin partition(datestr&＃61;&＃39;20210617&＃39;);

这里注意路径是我们刚才上传mapreduce程序执行结果文件的路径&＃xff0c;路径前后不要出现空格

查询

select * from ods_weblog_origin;

这段指令执行后我们可以看到的结果如下

创建明细表ods_weblog_detail

hive > create table ods_weblog_detail( valid string, --有效标识 remote_addr string, --来源IP remote_user string, --用户标识 time_local string, --访问完整时间 daystr string, --访问日期 timestr string, --访问时间 month string, --访问月 day string, --访问日 hour string, --访问时 request string, --请求的url status string, --响应码 body_bytes_sent string, --传输字节数 http_referer string, --来源url ref_host string, --来源的host ref_path string, --来源的路径 ref_query string, --来源参数query ref_query_id string, --来源参数query值 http_user_agent string --客户终端标识 ) partitioned by(datestr string); 18、创建中间临时表t_ods_tmp_referurl hive > create table t_ods_tmp_referurl as SELECT a.*,b.* FROM ods_weblog_origin a LATERAL VIEW parse_url_tuple(regexp_replace(http_referer, "\\"", ""), &＃39;HOST&＃39;, &＃39;PATH&＃39;,&＃39;QUERY&＃39;, &＃39;QUERY:id&＃39;) b as host, path, query, query_id;

创建临时中间表t_ods_tmp_detail

hive > create table t_ods_tmp_detail as select b.*,substring(time_local,0,10) as daystr, substring(time_local,12) as tmstr, substring(time_local,6,2) as month, substring(time_local,9,2) as day, substring(time_local,11,3) as hour from t_ods_tmp_referurl b;

加载数据到明细宽表前启用动态分区

hive > set hive.exec.dynamic.partition&＃61;true; hive > set hive.exec.dynamic.partition.mode&＃61;nonstrict;

生成明细宽表向ods_weblog_detail表&＃xff0c;加载数据

hive > insert overwrite table ods_weblog_detail partition(datestr) select distinct otd.valid,otd.remote_addr,otd.remote_user, otd.time_local,otd.daystr,otd.tmstr,otd.month,otd.day,otd.hour, otr.request,otr.status,otr.body_bytes_sent, otr.http_referer,otr.host,otr.path, otr.query,otr.query_id,otr.http_user_agent,otd.daystr from t_ods_tmp_detail as otd,t_ods_tmp_referurl as otr where otd.remote_addr&＃61;otr.remote_addr and otd.time_local&＃61;otr.time_local and otd.body_bytes_sent&＃61;otr.body_bytes_sent and otd.request&＃61;otr.request;

查看HDFS的WEB UI界面的ods_weblog_detail文件夹

统计每一天的PV量
# 创建表dw_pvs_everyday

hive > create table dw_pvs_everyday(pvs bigint,month string,day string);

#提取“day”字段

hive > insert into table dw_pvs_everyday select count(*) as pvs,owd.month as month,owd.day as day from ods_weblog_detail owd group by owd.month,owd.day;

查看表dw_pvs_everyday中的数据

Select * from dw_pvs_everyday;

实现人均浏览量
创建维度表dw_avgpv_user_everyday

hive > create table dw_avgpv_user_everyday( day string,avgpv string);

向表dw_avgpv_user_everyday中插入数据

hive > insert into table dw_avgpv_user_everyday select &＃39;2013-09-18&＃39;,sum(b.pvs)/count(b.remote_addr) from (select remote_addr,count(1) as pvs from ods_weblog_detail where datestr&＃61;&＃39;2013-09-18&＃39; group by remote_addr) b;

查看表dw_avgpv_user_everyday中的数据

Select * from dw_avgpv_user_everyday;

数据导出

通过SQLyog工具远程连接集群主服务器的MySQL服务

这里要注意的是链连接数据库出现Access denied for user ‘root‘
补充说明&＃xff1a;当别的机子&＃xff08;IP &＃xff09;通过客户端的方式在没有授权的情况下是无法连接 MySQL 数据库的&＃xff0c;如果需要远程连接 Linux 系统上的 MySQL 时&＃xff0c;必须为其 IP 和具体用户进行授权。一般 root 用户不会提供给开发者。如&＃xff1a;使用 Windows 上的 SQLyog 图形化管理工具连接 Linux 上的 MySQL 数据库&＃xff0c;必须先对其进行授权。
解决方法的参考链接https://blog.csdn.net/aotongkeji/article/details/123155896
然后测试连接

然后我们可以右击创建数据库

也可以

CREATE DATABASE if NOT EXISTS sqoopdb;

如图所在空白输入代码执行操作

创建七日人均浏览量表t_avgpv_num

mysql > create table &＃96;t_avgpv_num&＃96; ( &＃96;dateStr&＃96; varchar(255) DEFAULT NULL, &＃96;avgPvNum&＃96; decimal(6,2) DEFAULT NULL ) ENGINE&＃61;MyISAM DEFAULT CHARSET&＃61;utf8;

Sqoop导出数据
这一步是将我们的数据加载到mysql数据库中

sqoop export \\ --connect jdbc:mysql://192.168.80.140:3306/sqoopdb \\ --username hive\\ --password hive\\ --table t_avgpv_num \\ --columns "dateStr,avgPvNum" \\ --fields-terminated-by &＃39;\\001&＃39; \\ --export-dir /user/hive/warehouse/weblog.db/dw_avgpv_user_everyday;